package com.shanks.framework.mq.demo.config;

import com.shanks.framework.mq.data.mybatis.config.MybatisMQDataConfiguration;
import com.shanks.framework.mq.demo.constant.MQConstant;
import com.shanks.framework.mq.demo.processor.DemoProcessor;
import com.shanks.framework.mq.exception.MQException;
import com.shanks.framework.mq.producer.BaseMQProducer;
import com.shanks.framework.mq.rocketmq.config.RocketMQConfiguration;
import com.shanks.framework.mq.rocketmq.consumer.RocketMQConsumerAttribute;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;

@SpringBootConfiguration
@Import(MybatisMQDataConfiguration.class)
public class RocketMQConfig extends RocketMQConfiguration {

    /**
     * 配置生产者
     * @return
     * @throws MQException
     */
    @Bean
    public BaseMQProducer defaultRocketMQProducer() throws MQException {
        return createAndRegisterMQProducer(MQConstant.DEFAULT_PRODUCER, null);
    }

    /**
     * 配置生产者、消费者绑定订阅关系
     * @return
     * @throws MQException
     */
    @Bean
    @ConditionalOnProperty(name = "spring.rocketMQConsumer.enable", havingValue = "true")
    public DefaultMQPushConsumer defaultMQPushConsumer() throws MQException {
        /**
         * 配置消费者并且绑定消费关系
         */
        RocketMQConsumerAttribute rocketMQConsumerAttribute = newConsumerAttribute(MQConstant.DEFAULT_CONSUMER, null);
        rocketMQConsumerAttribute.subscribe(MQConstant.TOPIC.TOPIC_CHANGE_ORDER_STATUS, MQConstant.TAG.TAG_CHANGE_ORDER_STATUS, demoProcessor());

        /**
         * 生产者方配置消息确认topic
         */
        rocketMQConsumerAttribute.subscribeAckMsg(MQConstant.TOPIC.TOPIC_CHANGE_ORDER_STATUS, MQConstant.TAG.TAG_CHANGE_ORDER_STATUS);
        return createAndStartMQConsumer(rocketMQConsumerAttribute);
    }

    /**
     * 消费者处理器注入
     * @return
     * @throws MQException
     */
    @Bean
    public DemoProcessor demoProcessor() throws MQException{
        return new DemoProcessor(defaultRocketMQProducer());
    }
}
